package com.example.mysql.debezium.framework;

import io.debezium.embedded.EmbeddedEngine;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DebeziumEngineBean implements ApplicationContextAware, InitializingBean, DisposableBean {
    private final DebeziumEngine<ChangeEvent<String, String>> engine;
    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    public DebeziumEngineBean(Properties props) {
        // 2. 构建 DebeziumEngine
        // 使用 Json 格式
        EmbeddedEngine.EngineBuilder builder = new EmbeddedEngine.EngineBuilder();

        this.engine = DebeziumEngine.create(Json.class)
            .using(props)
            .notifying(this::handle)
            .using((success, message, error) -> {
                // 强烈建议加上此部分的回调代码，方便查看错误信息
                if (!success && error != null) {
                    // 报错回调
                    System.out.println("----------error------");
                    System.out.println(message);
                }
            }).build();
    }

    protected void handle(ChangeEvent<String, String> event) {
        // record中会有操作的类型（增、删、改）和具体的数据
        // key是主键
        System.out.println("record.key() = " + event.key());
        System.out.println("record.value() = " + event.value());
    }

    @Override
    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        Set<TableListener> listeners = new HashSet<>(ctx.getBeansOfType(TableListener.class).values());
        BinlogDispatcherListener binlogListener = new BinlogDispatcherListener(listeners);
        // this.client.registerEventListener(binlogListener);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        executor.submit(engine);
    }

    @Override
    public void destroy() throws Exception {
        executor.shutdown();
    }
}
